package org.example.flink.windows;

import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

public class Hello03CountWindowAll {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment environment = StreamExecutionEnvironment.getExecutionEnvironment();
        DataStreamSource<String> source = environment.socketTextStream("localhost", 19523);

        // Tumbling Window(滚动窗口)
//        source.map(word -> Tuple2.of(word.split(":")[0], Integer.parseInt(word.split(":")[1])),
//                        Types.TUPLE(Types.STRING, Types.INT))
//                .countWindowAll(3)
//                .reduce((t1, t2) -> {
//                    t1.f0 = t1.f0 + "_" +t2.f0;
//                    t1.f1 = t1.f1 + t2.f1;
//                    return t1;
//                }).print("CountWindow--Sliding:").setParallelism(1);


        //Sliding Window(滑动窗口)
        source.map(word -> Tuple2.of(word.split(":")[0], Integer.parseInt(word.split(":")[1])),
                        Types.TUPLE(Types.STRING, Types.INT))
                .countWindowAll(3, 2)
                .reduce((t1, t2) -> {
                    t1.f0 = t1.f0 + "_" +t2.f0;
                    t1.f1 = t1.f1 + t2.f1;
                    return t1;
                }).print("CountWindow--Sliding:").setParallelism(1);

        environment.execute();
    }
}
